package com.fwmagic.flink.cep;

import com.fwmagic.flink.pojo.Event;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

public class CEPExample {

    private static final Logger log = LoggerFactory.getLogger(CEPExample.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataSource = env.socketTextStream("localhost", 9898);

        //将socket数据流转换成Event
        SingleOutputStreamOperator<Event> eventDataStream = dataSource.flatMap(new FlatMapFunction<String, Event>() {
            @Override
            public void flatMap(String s, Collector<Event> collector) throws Exception {
                if (StringUtils.isNotBlank(s)) {
                    String[] split = s.split(",");
                    if (split.length == 2) {
                        collector.collect(new Event(Integer.valueOf(split[0]), String.valueOf(split[1])));
                    }
                }
            }
        });

        /**
         * 可以指定更多的 Pattern，通过指定的不同的连接条件
         *
         * next()：要求比较严格，该事件一定要紧跟着前一个事件。
         *
         * followedBy()：该事件在前一个事件后面就行，两个事件之间可能会有其他的事件。
         *
         * followedByAny()：该事件在前一个事件后面的就满足条件，两个事件之间可能会有其他的事件，返回值比上一个多。
         *
         * notNext()：不希望前一个事件后面紧跟着该事件出现。
         *
         * notFollowedBy()：不希望后面出现该事件。
         */
        //定义CEP匹配规则
        Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) throws Exception {
                log.info("start:{}", event.getId());
                return event.getId() == 42;
            }
        })./*next*//*followedBy*/followedByAny("middle").where(new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) throws Exception {
                log.info("middle:{}", event.getId());
                return event.getId() > 10;
            }
        });


        //CEP匹配规则触发
        CEP.pattern(eventDataStream, pattern).select(new PatternSelectFunction<Event, String>() {
            @Override
            public String select(Map<String, List<Event>> list) throws Exception {
                StringBuffer sb = new StringBuffer();
                log.info("list:{}",list);
                sb.append(list.get("start").get(0).getId())
                        .append(",").append(list.get("start").get(0).getName())
                        .append("\n")
                        .append(list.get("middle").get(0).getId()).append(",")
                        .append(list.get("middle").get(0).getName());
                return sb.toString();
            }
        }).print();

        env.execute("CEP example!");
    }
}
